package com.sqllog.messageMutipyThread;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sqllog.config.CanalClientConfig;
import com.sqllog.entity.RowChangeInfo;
import com.sqllog.listener.CanalEventListener;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public class MessageManagerResource {
    // canal 连接器
    private CanalConnector connector;

    private CanalClientConfig serverInstance;

    private String destination;

    private List<CanalEventListener> listenerArrays;

    // 仓库存储的载体
    private LinkedBlockingQueue<Message> messageLinkedBlockingQueue;

    public MessageManagerResource(CanalConnector connector, CanalClientConfig serverInstance, String destination, LinkedBlockingQueue<Long> errorMessageQueue, List<CanalEventListener> listenerArrays) {
        this.connector = connector;
        this.serverInstance = serverInstance;
        this.destination = destination;
        this.listenerArrays = listenerArrays;
        this.messageLinkedBlockingQueue = new LinkedBlockingQueue<>(serverInstance.getConsumerThreadSize());
    }

    /**
     * 生产资源
     */
    public void productMessage() throws Exception {
        //生产一个
        Message message = this.connector.getWithoutAck(this.serverInstance.getBatchSize());
        if (message.getId() == -1 || message.getEntries().size() == 0) {
            //暂停最多1秒
            Thread.sleep(new Random().nextInt(1000));
        }

        this.messageLinkedBlockingQueue.put(message);
    }

    public void consumerErrorMessage(){
        log.debug("当前消息队列大小 " + this.messageLinkedBlockingQueue.size());
    }

    /**
     * 处理消息
     * @throws InterruptedException
     */
    public void consumerMessage() throws InterruptedException{
        Message message = this.messageLinkedBlockingQueue.take();
        long messageId = message.getId();
        //实际消息数
        int size = message.getEntries().size();
        //若是没有消息
        if (messageId == -1 || size == 0) {
            //暂停最多1秒
            Thread.sleep(new Random().nextInt(1000));
        } else {
            try {
                if (Objects.isNull(this.connector)){
                    return;
                }

                // 进行应答，确认消息已被处理完
                this.connector.ack(messageId);
            }catch (NullPointerException e) {
                log.error("连接已丢失");
                return;
            }
            long currentTime = System.currentTimeMillis();
            this.analyzeMessage(message);
            log.debug("队列大小" + message.getEntries().size() + "消费消息消耗时间 " + (System.currentTimeMillis() - currentTime));
        }
    }


    public void analyzeMessage(Message message) {
        //获取 canal 消息处理的操作实体
        List<CanalEntry.Entry> messageList = message.getEntries();
        messageList.parallelStream().forEach(messageObj -> {
            //忽略不需要处理的消息的类型
            List<CanalEntry.EntryType> ignoreEntryTypes = getIgnoreEntryTypes();
            if (checkIsIgnoreEntryType(ignoreEntryTypes, messageObj)) {
                return;
            }

            // 转换 传输的二进制数组为可读的 rowChange 实体
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(messageObj.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
            // 通过实现类的方式处理
            for (CanalEventListener canalEventListener : this.listenerArrays) {
                RowChangeInfo rowChangeInfo = new RowChangeInfo(this.serverInstance.getHost(), this.destination, messageObj.getHeader().getSchemaName(), messageObj.getHeader().getTableName(), rowChange);
                canalEventListener.onEvent(rowChangeInfo);
            }
        });
    }

    public List<CanalEntry.EntryType> getIgnoreEntryTypes() {
        return Arrays.asList(CanalEntry.EntryType.TRANSACTIONBEGIN, CanalEntry.EntryType.TRANSACTIONEND, CanalEntry.EntryType.HEARTBEAT);
    }


    private boolean checkIsIgnoreEntryType(List<CanalEntry.EntryType> ignoreEntryTypes, CanalEntry.Entry messageObj){
        return CollectionUtil.isNotEmpty(ignoreEntryTypes) && ignoreEntryTypes.contains(messageObj.getEntryType());
    }
}